package net.bwie.realtime.jtp.dws.douyin.log.functions;
import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.FanRelation1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.LiveStatMetric1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.PayDetail1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.LiveUnifiedData1;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
public class LiveStatWindowFunction1 implements WindowFunction<
            LiveUnifiedData1,          // 输入：统一数据载体
            LiveStatMetric1,           // 输出：直播指标
            Long,                      // 分组键：主播ID（anchorId）
            TimeWindow> {              // 时间窗口

        @Override
        public void apply(
                Long anchorId,                  // 分组键：主播ID
                TimeWindow window,              // 窗口信息
                Iterable<LiveUnifiedData1> input, // 窗口内所有数据
                Collector<LiveStatMetric1> out) throws Exception {

            // 1. 初始化统计容器（按指标分类，与图片需求对齐）
            // 1.1 成交相关统计
            double totalDealAmount = 0.0;               // 累计成交金额（double）
            long totalDealGoodsCount = 0L;              // 成交件数（假设支付表order_id对应1件商品，实际需关联订单表）
            Set<Long> dealUserSet = new HashSet<>();    // 成交用户ID（去重）
            Set<Long> dealFanUserSet = new HashSet<>(); // 成交粉丝用户ID（去重）

            // 1.2 观看/在线相关统计
            Set<Long> viewUserSet = new HashSet<>();     // 累计观看用户ID（去重）
            long totalOnlineUserSum = 0L;               // 实时在线人数总和（用于计算平均）
            int onlineRecordCount = 0;                  // 在线人数记录次数（用于计算平均）

            // 1.3 粉丝相关统计
            Set<Long> newFanUserSet = new HashSet<>();  // 新增粉丝用户ID（去重）
            Set<Long> newFanClubUserSet = new HashSet<>(); // 新加粉丝团用户ID（去重）

            // 1.4 时长相关统计
            long totalWatchDuration = 0L;               // 总看播时长（秒）

            // 2. 遍历窗口数据，按类型统计（通过dataTypeFlag区分数据来源）
            Iterator<LiveUnifiedData1> iterator = input.iterator();
            while (iterator.hasNext()) {
                LiveUnifiedData1 unifiedData = iterator.next();
                Integer flag = unifiedData.getDataTypeFlag();

                // 2.1 处理支付表数据（flag=1：统计成交金额、成交人数、成交件数）
                if (flag == 1 && unifiedData.getPayData() != null) {
                    PayDetail1 pay = unifiedData.getPayData();
                    if ("success".equals(pay.getPayStatus())) { // 仅统计成功支付
                        totalDealAmount += pay.getPayAmount(); // 累加成交金额（double）
                        totalDealGoodsCount++; // 累加成交件数（简化：1订单=1件，实际需关联订单表order_cnt）
                        dealUserSet.add(pay.getUserId()); // 记录成交用户

                        // 标记成交粉丝（后续通过粉丝关系表补充）
                        dealFanUserSet.add(pay.getUserId());
                    }
                }

                // 2.2 处理粉丝关系表数据（flag=2：统计新增粉丝、成交粉丝占比）
                else if (flag == 2 && unifiedData.getFanData() != null) {
                    FanRelation1 fan = unifiedData.getFanData();
                    // 统计新增粉丝（is_new=yes）
                    if ("yes".equals(fan.getIsNew()) && !newFanUserSet.contains(fan.getUserId())) {
                        newFanUserSet.add(fan.getUserId());
                    }
                    // 统计新加粉丝团（简化：假设is_new=yes且change_time在窗口内为新增粉丝团，实际需粉丝团标识）
                    if ("yes".equals(fan.getIsNew())) {
                        newFanClubUserSet.add(fan.getUserId());
                    }
                    // 补充成交粉丝判断（非粉丝不加入成交粉丝集合）
                    if ("yes".equals(fan.getIsNotFan())) {
                        dealFanUserSet.remove(fan.getUserId());
                    }
                }

                // 2.3 处理直播日志数据（flag=3：统计观看人数、在线人数、看播时长）
                else if (flag == 3 && unifiedData.getLogData() != null) {
                    EventLog6 log = unifiedData.getLogData();
                    // 统计累计观看人数（event_type=进入直播间）
                    if ("进入直播间".equals(log.getEventType())) {
                        viewUserSet.add(log.getUserId());
                    }
                    // 统计实时在线人数（简化：假设log.duration_sec为实时在线人数，实际需专用字段）
                    if ("在线人数统计".equals(log.getEventType())) {
                        totalOnlineUserSum += log.getDurationSec();
                        onlineRecordCount++;
                    }
                    // 统计看播时长（event_type=退出直播间，duration_sec为停留时长）
                    if ("退出直播间".equals(log.getEventType())) {
                        totalWatchDuration += log.getDurationSec();
                    }
                }
            }

            // 3. 计算核心指标（避免除零错误，比率保留4位小数，用double存储）
            long totalViewCount = viewUserSet.size();       // 累计观看人数
            long totalDealUserCount = dealUserSet.size();   // 成交人数
            long totalDealFanCount = dealFanUserSet.size(); // 成交粉丝人数

            // 3.1 成交转化率 = 成交人数 / 累计观看人数
            double dealConversionRate = totalViewCount == 0 ? 0.0 :
                    Math.round((totalDealUserCount * 1.0 / totalViewCount) * 10000) / 10000.0;

            // 3.2 千次观看成交金额 = (累计成交金额 / 累计观看人数) * 1000
            double perThousandViewAmount = totalViewCount == 0 ? 0.0 :
                    Math.round((totalDealAmount / totalViewCount) * 1000 * 100) / 100.0;

            // 3.3 成交粉丝占比 = 成交粉丝数 / 成交总人数
            double dealFanRatio = totalDealUserCount == 0 ? 0.0 :
                    Math.round((totalDealFanCount * 1.0 / totalDealUserCount) * 10000) / 10000.0;

            // 3.4 平均在线人数 = 在线人数总和 / 记录次数
            double avgOnlineUserCount = onlineRecordCount == 0 ? 0.0 :
                    Math.round((totalOnlineUserSum * 1.0 / onlineRecordCount) * 100) / 100.0;

            // 3.5 人均看播时长 = 总看播时长 / 累计观看人数（秒）
            double avgWatchDuration = totalViewCount == 0 ? 0.0 :
                    Math.round((totalWatchDuration * 1.0 / totalViewCount) * 100) / 100.0;

            // 3.6 新增粉丝数/新加粉丝团人数（去重后计数）
            long newFanCount = newFanUserSet.size();
            long newFanClubCount = newFanClubUserSet.size();

            // 4. 格式化窗口时间（与原有OrderSourceRatioWindowFunction3一致）
            String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
            String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
            String curDate = windowStartTime.substring(0, 10);

            // 5. 封装指标并输出（直播ID预留，可从日志表提取）
            LiveStatMetric1 metric = new LiveStatMetric1(
                    windowStartTime,
                    windowEndTime,
                    curDate,
                    anchorId,
                    "LIVE_" + anchorId + "_" + window.getStart(), // 生成临时直播ID
                    totalDealAmount,
                    totalDealGoodsCount,
                    totalDealUserCount,
                    dealConversionRate,
                    perThousandViewAmount,
                    dealFanRatio,
                    avgOnlineUserCount,
                    totalViewCount,
                    newFanClubCount,
                    newFanCount,
                    avgWatchDuration
            );
            out.collect(metric);
        }
}
